package org.example.com.RocketMq.smallCase;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * 同步发送消息-消费者
 *
 * @author qiaohaojie
 * @date 2023/2/20  21:39
 */
public class RocketMqConsumer {
    public static void main(String[] args) throws Exception {
        // 1. 创建DefaultMQPushConsumer实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group_consumer_demo1");
        // 2. 设置nameServer地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        // 设置消息拉取最大数
        consumer.setConsumeMessageBatchMaxSize(2);
        // 3. 设置subscribe，这里是要读取的主题信息
        consumer.subscribe("Topic_Demo", "*");
        // 4. 设置消息监听
        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                // 5. 获取消息信息
                // 迭代消息信息
                for (MessageExt msg : msgs) {
                    try {
                        // 获取主题
                        String topic = msg.getTopic();
                        // 获取标签
                        String tags = msg.getTags();
                        // 获取信息
                        String result = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);
                        System.out.println("Consumer消费消息--topic:" + topic + ",targs:" + tags + ",result:" + result);
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                        // 消息重试
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                // 6. 返回消息读取状态
                // 消息消费完成
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 开启Consumer
        consumer.start();
    }
}
